package com.linjiahao.nettyGateway.cluster;

import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.linjiahao.commonUtil.Redis.RedisService;
import com.linjiahao.nettyGateway.constant.Constant;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.locks.StampedLock;
import java.util.stream.Collectors;

/**
* 自定义用户连接netty存储信息
 * 才有本地哈希环+redis缓存用户连接节点信息
* */
@Slf4j
public class ClusterManager {

    @Resource
    private RedisService redisService;

    @Value("${netty.virtual.node}")
    private Integer VIRTUAL_NODES;


    private final StampedLock stampedLock = new StampedLock();

    private static SortedMap<Integer, ServerNode> virtualNodes = new TreeMap<Integer, ServerNode>();

    private static List<Instance> currentInstance = new ArrayList<>();

    public void refreshNettyServer(List<Instance> instances){
        log.info("刷新一致性hash环,服务数量为：{}",instances.size());
        if(CollectionUtils.isEmpty(instances)){
            //可用服务为空，自定义发送告警
        }
        if(!CollectionUtils.isEmpty(currentInstance) && instances.size()==currentInstance.size()){
            Instance instance = currentInstance.stream().filter(v -> !instances.stream().map(Instance::getInstanceId).collect(Collectors.toList()).contains(v.getInstanceId())).findFirst().orElse(null);
            if(instance==null){
                log.info("当前无实例进行变化");
                return;
            }
        }
        currentInstance = instances;
        long lock = stampedLock.writeLock();
        try {
            virtualNodes.clear();
            instances.stream().parallel().forEach(v->{
                for (Integer i = 0; i < VIRTUAL_NODES; i++) {
                    String virtualNodeName = v.getIp()+":"+v.getPort()+  "&&VN" + String.valueOf(i);
                    int hash = getHash(virtualNodeName);
                    ServerNode serverNode = ServerNode.builder().instanceId(v.getInstanceId())
                            .host(v.getIp()).port(v.getPort()).clusterName(v.getServiceName()).build();
                    virtualNodes.put(hash, serverNode);
                }
            });
        }catch (Exception e){
            log.error("刷新一致性hash环异常:{]",e);
        }finally {
            stampedLock.unlockWrite(lock);
            log.info("刷新一致性hash环结果为：{}",JSON.toJSONString(instances));
        }
    }

    private  int getHash(String str) {
        final int p = 16777619;
        int hash = (int)2166136261L;
        for (int i = 0; i < str.length(); i++) {
            hash = (hash ^ str.charAt(i)) * p;
        }
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;

        // 如果算出来的值为负数则取其绝对值
        if (hash < 0) {
            hash = Math.abs(hash);
        }
        return hash;
    }

    /**
     * 得到应当路由到的结点
     */
    public  ServerNode getServer(String node) {
        // 得到带路由的结点的Hash值
        int hash = getHash(node);
        // 得到大于该Hash值的所有Map
        long optimisticRead = stampedLock.tryOptimisticRead();
        SortedMap<Integer, ServerNode> subMap = virtualNodes;
        if(!stampedLock.validate(optimisticRead)){
            optimisticRead = stampedLock.readLock();
            try {
                subMap = virtualNodes;
            }finally {
                stampedLock.unlockRead(optimisticRead);
            }
        }
        if(!virtualNodes.isEmpty()){
            subMap = virtualNodes.tailMap(hash);
            // 第一个Key就是顺时针过去离node最近的那个结点
            Integer i = subMap.firstKey();
            // 返回对应的虚拟节点名称，这里字符串稍微截取一下
            ServerNode virtualNode = subMap.get(i);
            return virtualNode;
        }
        return null;
    }


    /*
    * 客户端上线处理
    * */
    public Channel addChannel(Channel channel,UserChannelInfo userChannelInfo) {
        //添加服务器下的用户
        redisService.hset(String.format(Constant.NETTY_CONNECT_USER_KEY,userChannelInfo.getServerHost()+userChannelInfo.getServerPort()),
                userChannelInfo.getUserId(),JSON.toJSONString(userChannelInfo));
        //添加缓存用户连接所在服务器
        redisService.hset(Constant.USER_CONNECT_KEY, userChannelInfo.getUserId(),JSON.toJSONString(userChannelInfo));
        return channel;

    }

    /**
     * 客户端下线处理
     * @param userChannelInfo
     */
    public void channelCloseHandle(UserChannelInfo userChannelInfo) {
        log.info("- - - - - - - - - " + userChannelInfo.getUserId() + " offline from server " + userChannelInfo.getServerHost() + "  - - - - - - - - - ");
        redisService.hdel(String.format(Constant.NETTY_CONNECT_USER_KEY,userChannelInfo.getServerHost()+userChannelInfo.getServerPort()),
                userChannelInfo.getUserId());
        redisService.hdel(Constant.USER_CONNECT_KEY,userChannelInfo.getUserId());
    }



}

